package com.vitalgateway.component.message.listener;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.dtflys.forest.annotation.SSEDataMessage;
import com.dtflys.forest.annotation.SSEName;
import com.dtflys.forest.annotation.SSEValue;
import com.dtflys.forest.http.ForestSSE;
import com.dtflys.forest.sse.EventSource;
import com.vitalgateway.dao.entity.AccountStatusEvent;
import com.vitalgateway.dao.repository.AccountStatusEventRepository;
import com.vitalgateway.object.dto.alpaca.MissingIdRange;
import com.vitalgateway.service.AlpacaService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.annotation.Lazy;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * SSE 事件处理器
 */
@Slf4j
@Service
@Order(10)
public class AlpacaPushSSE implements CommandLineRunner {

    //当前SSE连接
    protected static volatile ForestSSE currentSSE;
    private static AtomicLong nextEventId;
    @Resource
    private AccountStatusEventRepository accountStatusEventRepository;
    @Resource
    private AlpacaService alpacaService;
    @Resource
    @Lazy
    private AlpacaPushSSE alpacaPushSSE;

    /**
     * 处理业务逻辑
     */
    private void handleSSEMessage(AccountStatusEvent accountStatusEvent) {
        // 推送过来的消息的eventId
        Long eventId = accountStatusEvent.getEventId();
        log.info("AlpacaPushSSE#handleSSEMessage：eventId={}, nextEventId={}", eventId, nextEventId.get());

        if (Objects.equals(eventId, nextEventId.get())) {
            // 如果推送的eventId等于数据库中下一个应有的eventId
            accountStatusEventRepository.save(accountStatusEvent);
            nextEventId.incrementAndGet();
            log.info("AlpacaPushSSE#handleSSEMessage： saved message eventId={}", eventId);
        } else if (eventId > nextEventId.get()) {
            //如果当前接收到的 event_id > nextEventId，则认为中间有数据丢失,触发一次数据补偿，补偿的范围为  nextEventId到当前event_id；
            MissingIdRange missingIdRange = MissingIdRange.builder().startId(nextEventId.get()).endId(eventId).build();
            List<MissingIdRange> missingIdRanges = Collections.singletonList(missingIdRange);
            queryMissDataAndSave(missingIdRanges);
        } else {
            // event_id < nextEventId 检查该消息是否已存在
            int exists = accountStatusEventRepository.count(new LambdaQueryWrapper<AccountStatusEvent>().eq(AccountStatusEvent::getEventId, eventId));
            if (exists > 0) {
                log.warn("AlpacaPushSSE#handleSSEMessage：message eventId={} is exists,skip", eventId);
            } else {
                accountStatusEventRepository.save(accountStatusEvent);
                log.info("AlpacaPushSSE#handleSSEMessage：saved message eventId={}", eventId);
            }
        }
    }

    /**
     * 消息补偿 通过定时任务进行补偿
     */
    public void compensateMissingMessages() {
        try {
            log.info("AlpacaPushSSE#compensateMissingMessages: alpaca sse start compensate missing messages");
            Long sseMsgNextEventId = getSSEMsgNextEventId();
            if (sseMsgNextEventId <= 0) {
                return;
            }

            List<MissingIdRange> missingIdRanges = new ArrayList<>();
            // 每次查询 500,000 条
            int groupSize = 500000;
            for (long i = 0; i <= sseMsgNextEventId; i += groupSize) {
                long groupStart = i;
                long groupEnd = Math.min(i + groupSize - 1, sseMsgNextEventId);
                // 查询当前分区范围内的缺失 ID
                List<MissingIdRange> partitionMissingIdRanges = accountStatusEventRepository.findMissingIdRanges(groupStart, groupEnd);
                // 合并查询结果到集合中
                if (!CollectionUtils.isEmpty(partitionMissingIdRanges)) {
                    missingIdRanges.addAll(partitionMissingIdRanges);
                }
            }
            queryMissDataAndSave(missingIdRanges);
            log.info("AlpacaPushSSE#compensateMissingMessages: alpaca sse complete compensate missing messages");
        } catch (Exception e) {
            log.warn("AlpacaPushSSE#compensateMissingMessages error :{}", e.getMessage(), e);
        }
    }

    /**
     * 查询缺失的数据并且补齐
     *
     * @param missingIdRanges 缺失的范围
     */
    private void queryMissDataAndSave(List<MissingIdRange> missingIdRanges) {
        if (CollectionUtils.isEmpty(missingIdRanges)) {
            log.info("AlpacaPushSSE#queryMissDataAndSave: no missing ID range is found");
            return;
        }

        log.info("AlpacaPushSSE#queryMissDataAndSave: {}", missingIdRanges);
        for (MissingIdRange missingIdRange : missingIdRanges) {
            Long missingStart = missingIdRange.getStartId();
            Long missingEnd = missingIdRange.getEndId();
            log.info("AlpacaPushSSE#queryMissDataAndSave: subscribe missing messages startId={}, endId={}", missingStart, missingEnd);
            subscribeMissingMessages(missingStart, missingEnd, new AlpacaPushSSE());
        }
    }

    /**
     * 获取sse消息应该推送的下一条消息的eventId
     * 若没有数据 从0开始，有数据从max EventId+1
     *
     * @return next eventId
     */
    private Long getSSEMsgNextEventId() {
        AccountStatusEvent sseMsg = accountStatusEventRepository.getOne(new LambdaQueryWrapper<AccountStatusEvent>().orderByDesc(AccountStatusEvent::getEventId).last("limit 1"));
        return Optional.ofNullable(sseMsg).map(AccountStatusEvent::getEventId).map(eventId -> eventId + 1).orElse(0L);
    }


    /**
     * @param start 起始位置
     * @param end   结束位置
     * @return 缺失的数据
     */
    private void subscribeMissingMessages(Long start, Long end, AlpacaPushSSE sse) {
        // 如果 start 大于 end
        if (start > end) {
            return;
        }
        // 订阅
        try {
            alpacaService.subscribeToAccountStatus(null, null, start, end, null, null, sse);
        } catch (Exception e) {
            log.error("AlpacaPushSSE#subscribeMissingMessages error: {}", e.getMessage(), e);
        }
    }


    /**
     * 订阅
     */
    public void createSSE() {
        if (currentSSE != null) {
            return;
        }
        //获取应获取eventId的起始位置
        Long eventId = getSSEMsgNextEventId();
        nextEventId = new AtomicLong(eventId);

        try {
            alpacaService.subscribeToAccountStatus(null, null, eventId, null, null, null, alpacaPushSSE);
        } catch (Exception e) {
            log.error("AlpacaPushSSE#createSSE error: {}", e.getMessage(), e);
        }
    }


    @Override
    public void run(String... args) throws Exception {
        createSSE();
    }

    public void onOpen(EventSource eventSource) {
        currentSSE = eventSource.sse();
        log.info("alpaca SSE onOpen");
    }

    public void onClose(EventSource eventSource) {
        log.info("alpaca SSE onClose");
        currentSSE = null;
    }

    @SSEDataMessage
    public void onSSEData(EventSource eventSource, @SSEName String name, @SSEValue String value) {
        log.info("alpaca SSE Push Message Content:{}", value);
        AccountStatusEvent accountStatusEvent = JSONObject.parseObject(value, AccountStatusEvent.class);
        handleSSEMessage(accountStatusEvent);
    }
}